package edu.ucsd.db.bassarisk.algebra.logical.visitor;

import java.util.ArrayList;
import java.util.List;

import edu.ucsd.db.bassarisk.algebra.logical.AggregateAll;
import edu.ucsd.db.bassarisk.algebra.logical.AggregateWithin;
import edu.ucsd.db.bassarisk.algebra.logical.ConstantTuple;
import edu.ucsd.db.bassarisk.algebra.logical.Intersection;
import edu.ucsd.db.bassarisk.algebra.logical.Join;
import edu.ucsd.db.bassarisk.algebra.logical.JoinWithin;
import edu.ucsd.db.bassarisk.algebra.logical.LogicalOperator;
import edu.ucsd.db.bassarisk.algebra.logical.Map;
import edu.ucsd.db.bassarisk.algebra.logical.Scan;
import edu.ucsd.db.bassarisk.algebra.logical.Select;
import edu.ucsd.db.bassarisk.algebra.logical.Union;
import edu.ucsd.db.bassarisk.algebra.logical.WithinContext;
import edu.ucsd.db.bassarisk.algebra.logical.function.Function;
import edu.ucsd.db.bassarisk.query.binding.BindingTuple;
import edu.ucsd.db.bassarisk.query.binding.Bindings.Binding;

public class ComputeOutputDescriptors implements LogicalOperatorVisitor<BindingTuple, Void> {

  @Override
  public BindingTuple visitAggregateAll(AggregateAll op, Void arg) {
    op.subPlan().accept(this, arg);
    BindingTuple tuple = new BindingTuple();
    for (Binding binding : op.aggregates()) {
      tuple.addField(binding);
    }
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitAggregateWithin(AggregateWithin op, Void arg) {
    op.subPlan().accept(this, arg);
    BindingTuple tuple = new BindingTuple();
    for (Binding binding : op.aggregates()) {
      tuple.addField(binding);
    }
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitConstantTuple(ConstantTuple op, Void arg) {
    BindingTuple tuple = new BindingTuple();
    for (Function f : op.fields()) {
      tuple.addField(f.getBinding());
    }
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitIntersection(Intersection op, Void arg) {
    BindingTuple tuple = appendOutputDescriptors(op.intersectionPlans());
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitJoin(Join op, Void arg) {
    BindingTuple tuple = appendOutputDescriptors(op.joinInputs());
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitJoinWithin(JoinWithin op, Void arg) {
    BindingTuple tuple = appendOutputDescriptors(op.joinInputs());
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitMap(Map op, Void arg) {
    BindingTuple tuple = new BindingTuple();
    BindingTuple subTuple = op.subPlan().accept(this, null);
    for (Binding binding : subTuple.bindings()) {
      tuple.addField(binding);
    }
    for (Binding binding : op.functions()) {
      tuple.addField(binding);
    }
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitSelect(Select op, Void arg) {
    BindingTuple tuple = op.subPlan().accept(this, arg);
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitScan(Scan op, Void arg) {
    BindingTuple tuple = new BindingTuple();
    tuple.addField(op.getBindingSatisfied());
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitUnion(Union op, Void arg) {
    BindingTuple tuple = appendOutputDescriptors(op.unionPlans());
    op.setOutputDesciptor(tuple);
    return tuple;
  }

  @Override
  public BindingTuple visitWithinContext(WithinContext op, Void arg) {
    BindingTuple tuple = op.subPlan().accept(this, arg);
    op.setOutputDesciptor(tuple);
    return tuple;
  }
  
  private BindingTuple appendOutputDescriptors(Iterable<LogicalOperator> subPlans) {
    List<Binding> bindings = new ArrayList<Binding>();
    for (LogicalOperator op : subPlans) {
      BindingTuple subTuple = op.accept(this, null);
      for (Binding binding : subTuple.bindings()) {
        bindings.add(binding);
      }
    }
    return new BindingTuple(bindings);
  }
}
